use crate::workflows::workflow::Workflow;
use crate::error::ServerError;
use crate::abstractions::data::Data;

use super::manager::WORKFLOW_MANAGER;
use super::workflow::WorkflowRunningStatus;

pub fn manager_update(id: u16, progress: &Vec<(String, u64)>, status: &WorkflowRunningStatus) {
    let mut manager = WORKFLOW_MANAGER.get().unwrap().lock().unwrap();
    manager.update_portable_running_workflow(id, progress.clone(), status.clone());
    drop(manager);
}

pub fn manager_update_end(id: u16, progress: &Vec<(String, u64)>, status: &WorkflowRunningStatus, output: Result<Data, ServerError>) {
    let mut manager = WORKFLOW_MANAGER.get().unwrap().lock().unwrap();
    manager.update_portable_running_workflow(id, progress.clone(), status.clone());
    manager.protable_running_result_in(id, output);
    drop(manager);
}

pub fn start_workflow_func(workflow: &Workflow, input_data: String, port: u16) -> Result<(), ServerError> {
    let json_value = serde_json::from_str::<serde_json::Value>(&input_data).map_err(|e| ServerError::WorkflowError(e.to_string()))?;
    let input = Data::try_from_json_less(json_value, workflow.tin())?;
    match workflow {
        Workflow::ChainPortable(t, given_name) => {
            let name_main = given_name.clone();
            let fun = t.sendfun().clone();
            let handle = std::thread::spawn(move || {
                let output = fun(input, port);
                // println!("工作流线程退出: {}, {:?}", name_main, output);
                output
            });
            let signals = t.signals().clone();
            let main_serial_id = t.serial_id();
            std::thread::spawn(move || {
                // zmq监听
                let context = zmq::Context::new();
                let socket = context.socket(zmq::PULL).unwrap();
                let endpoint = format!("tcp://127.0.0.1:{}", port);
                socket.bind(&endpoint).unwrap();
                let min_interval = std::time::Duration::from_millis(500);
                let mut status = WorkflowRunningStatus::Running(0);
                let mut progress = vec![];
                let mut current_task_idx = 0;
                let mut current_serial_id = 0;
                if signals.len() > 0 {
                    current_task_idx = 0;
                    current_serial_id = signals[current_task_idx].0;
                    let current_task_name = signals[current_task_idx].1.clone();
                    progress.push((current_task_name, 0));
                }
                manager_update(port, &progress, &status);
                let mut last_update_time = std::time::Instant::now();
                loop {
                    let bs = socket.recv_bytes(0).unwrap();
                    let serial_id = std::str::from_utf8(&bs).unwrap().parse::<u64>().unwrap();
                    println!("收到: {}", serial_id);
                    if serial_id == current_serial_id {
                        progress[current_task_idx].1 += 1;
                    } else {
                        if signals.len() == 0 || current_task_idx >= signals.len() - 1 {
                            if serial_id == main_serial_id {
                                status = WorkflowRunningStatus::Finished;
                                let output = handle.join().unwrap();
                                println!("workflow finished: {:?}", output);
                                if let Err(e) = write_to_file(name_main.clone(), &progress, &status, &output) {
                                    eprintln!("写入文件失败: {:?}", e);
                                }
                                manager_update_end(port, &progress, &status, output);
                                break;
                            } else {
                                // status = WorkflowRunningStatus::Failed(format!("最终信号不正确, 期望{} 收到{}", main_serial_id, serial_id));
                                status = WorkflowRunningStatus::Finished;
                                let output = handle.join().unwrap();
                                println!("workflow failed: {:?}", output);
                                if let Err(e) = write_to_file(name_main.clone(), &progress, &status, &output) {
                                    eprintln!("写入文件失败: {:?}", e);
                                }
                                manager_update_end(port, &progress, &status, output);
                                break;
                            }
                        }
                        current_task_idx += 1;
                        current_serial_id = signals[current_task_idx].0;
                        // if serial_id != current_serial_id {
                        //     status = WorkflowRunningStatus::Failed(format!("任务信号不正确, 期望{} 收到{}", current_serial_id, serial_id));
                        //     manager_update(port, &progress, &status);
                        //     break;
                        // }
                        let current_task_name = signals[current_task_idx].1.clone();
                        progress.push((current_task_name, 1));
                    }
                    if last_update_time.elapsed() > min_interval {
                        manager_update(port, &progress, &status);
                        last_update_time = std::time::Instant::now();
                    }
                }
            });
            Ok(())
        }
        _ => Err(ServerError::WorkflowError(format!("工作流没有融合: {}", workflow.name())))
    }

}

pub fn write_to_file(name: String, progress: &Vec<(String, u64)>, status: &WorkflowRunningStatus, output: &Result<Data, ServerError>) -> Result<(), ServerError> {
    let file_name = format!("edp_files/workflows/{}-{}.json", name, crate::prelude::now_str());
    // 创建文件夹
    let file_dir = std::path::Path::new(&file_name).parent().unwrap();
    std::fs::create_dir_all(file_dir).unwrap();
    let file = std::fs::File::create(file_name).unwrap();
    let writer = std::io::BufWriter::new(file);
    match output {
        Ok(data) => {
            let res = serde_json::json!({
                "name": name,
                "time": crate::prelude::now_str(),
                "progress": progress,
                "status": status,
                "output": data.to_json_less(),
            });
            serde_json::to_writer(writer, &res).map_err(|e| ServerError::WorkflowError(e.to_string()))?;
            Ok(())
        }
        Err(e) => {
            let res = serde_json::json!({
                "name": name,
                "time": crate::prelude::now_str(),
                "progress": progress,
                "status": status,
                "error": format!("{:?}", e),
            });
            serde_json::to_writer(writer, &res).map_err(|e| ServerError::WorkflowError(e.to_string()))?;
            Ok(())
        }
    }
}

pub fn get_available_port() -> u16 {
    static PORT_AVAILABLE: std::sync::atomic::AtomicU16 = std::sync::atomic::AtomicU16::new(20000);
    let mut port = PORT_AVAILABLE.fetch_add(0, std::sync::atomic::Ordering::Relaxed);
    loop {
        if !is_port_in_use(port) {
            return port;
        }
        port = PORT_AVAILABLE.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    }
}

fn is_port_in_use(port: u16) -> bool {
    let socket = std::net::TcpListener::bind(("0.0.0.0", port));
    match socket {
        Ok(_) => false,
        Err(_) => true,
    }
}

